package org.ala.tiktools.rocketmq.comsumer;


import org.ala.tiktools.logs.LoggerFactory;
import org.ala.tiktools.rocketmq.MQMessage;
import org.ala.tiktools.rocketmq.commons.BizThreadPoolExecutor;
import org.ala.tiktools.rocketmq.serializable.kryo.KryoTool;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;

import java.lang.reflect.Method;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicInteger;


/**
 * consumer方法执行器
 *
 * @author ala
 * @date 2024-11-29 14:01
 */
public class ConsumerExecutor {


    static Logger log = LoggerFactory.getLogger("rocketmq", "executor");


    /**
     *  group, topic, tag
     */
    protected String group, topic, tag;
    /**
     *  执行线程池
     */
    protected BizThreadPoolExecutor threadPool;
    /**
     *  执行方法的bean
     */
    protected Object bean;
    /**
     *  要执行的method
     */
    protected Method method;
    /**
     *  最大积压任务数
     */
    protected int queueSize;
    /**
     *  恢复消费阈值
     */
    protected int threshold;
    /**
     *  Consumer
     */
    protected DefaultMQPushConsumer consumer;
    /**
     *  被堆积待消费的消息数
     */
    protected AtomicInteger backlog = new AtomicInteger(0);


    /**
     *  构建ConsumerExecutor
     */
    public static ConsumerExecutor build(String beanName, Object bean, Method method,
                                         BizThreadPoolExecutor threadPool,
                                         String namesrvAddrs, String group, String topic, String tag,
                                         boolean boardCasting, boolean orderly) throws MQClientException {
        ConsumerExecutor executor = new ConsumerExecutor();
        executor.method = method;
        executor.bean = bean;

        //  给线程池
        executor.threadPool = threadPool;

        //  初始化Consumer
        executor.group = group;
        executor.topic = topic;
        executor.tag = tag;
        executor.initConsumer(namesrvAddrs, group, topic, tag,
                boardCasting, orderly);

        return executor;
    }
    protected void initConsumer(String namesrvAddrs,
                                String group, String topic, String tag,
                                boolean boardCasting, boolean orderly) throws MQClientException {
        consumer = new DefaultMQPushConsumer(group);
        if (orderly) {
            consumer.setMessageListener(new MessageListenerOrderly() {
                @Override
                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                    msgs.forEach(ext -> {
                        invokeMsg(ext);
                    });
                    return ConsumeOrderlyStatus.SUCCESS;
                }
            });
        } else {
            consumer.setMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    msgs.forEach(ext -> {
                        invokeMsg(ext);
                    });
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
        }
        if (boardCasting) {
            consumer.setMessageModel(MessageModel.BROADCASTING);
        } else {
            consumer.setMessageModel(MessageModel.CLUSTERING);
        }
//        consumer.off
        consumer.setNamesrvAddr(namesrvAddrs);
        consumer.subscribe(topic, tag);
        consumer.start();

        log.info("[ConsumerExecutor initConsumer] 开启consumer消费 namesrc:" + namesrvAddrs + " topic:" + topic + " group:" + group + " tag:" + tag);
    }


    /**
     *  执行消费方法
     */
    protected void invokeMsg(MessageExt msgExt) {
        threadPool.execute(
                new Runnable() {
                        @Override
                        public void run() {
                            try {
                                doInvork(msgExt);
                            } catch (Exception e) {
                                throw new RuntimeException(e);
                            }
                        }
                    },
                new BizThreadPoolExecutor.IBacklog() {
                    @Override
                    public void onSubmit(int backlog) {
                        //  如果积压消息数量超过队列容量，暂停消费
                        if (backlog >= queueSize && !consumer.isPause()) {
                            consumer.suspend();
                        }
                    }

                    @Override
                    public void onComplate(int backlog) {
                        //  如果积压消息数降到阈值一下，恢复消费
                        if (backlog <= threshold && consumer.isPause()) {
                            consumer.resume();
                        }
                    }
                });
    }
    protected void doInvork(MessageExt msg) throws Exception {
        //  反序列化成要执行的参数
        MQMessage mqMessage = KryoTool.getInstance().deserialize(msg.getBody(), MQMessage.class);
        log.info("[ConsumerExecutor doInvork] 收到消息 group:" + group + " topic:" + topic + " tag:" + tag + " id:" + mqMessage.getId());

        method.invoke(bean, mqMessage.getBody());
    }

}
